package com.why.websocketmqttstudy.why.mqtt.callback;

import com.why.websocketmqttstudy.why.mqtt.client.MqttConsumerClient;
import com.why.websocketmqttstudy.why.mqtt.handle.HandlerFactory;
import com.why.websocketmqttstudy.why.mqtt.handle.PocMsgHandler;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;

/**
 * 主题消息回调
 * @author why
 *
 */
public class TopicMsgCallback implements MqttCallback {

//	private final static Logger logger = LoggerFactory.getLogger(TopicMsgCallback.class);
	
	private MqttConsumerClient mqttConsumerClient;
	private HandlerFactory handlerFactory;
	
    public TopicMsgCallback(MqttConsumerClient mqttConsumerClient) {
     this.mqttConsumerClient = mqttConsumerClient; 
     this.handlerFactory = mqttConsumerClient.getHandlerFactory();
    }

    /**
     * 失去连接，考虑重连
     */
	@Override
	public void connectionLost(Throwable cause) 
	{
//		logger.error("失去连接",cause);
		mqttConsumerClient.reConnect();		
	}

	/**
	 * 订阅消息到达
	 */
	@Override
	public void messageArrived(String topic, MqttMessage message) throws Exception 
	{
//		logger.info("Topic: " + topic);
        final String payload = new String(message.getPayload());
//        logger.info("payload: " + payload);
        
        // 根据主题获取注册的处理服务
        PocMsgHandler handler = handlerFactory.getHandler(topic);
        if(null == handler)
        {
//        	logger.error("未注册主题【"+topic+"】的处理服务");
        	return;
        }
        try{
			handler.handle(payload);
		}
        catch (Exception e){
//        	logger.error("请求失败  异常："+e);
		}
	}

	/**
	 * 消息投递成功
	 */
	@Override
	public void deliveryComplete(IMqttDeliveryToken token) {
		
	}
}
